package com.game.net.client.jpbsocket;

import com.game.net.BaseClient;
import com.game.message.Msg;
import com.game.message.jpb.codec.JPbDecoder;
import com.game.message.jpb.codec.JPbEncoder;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.logging.LoggingHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;


@Component
@Scope("prototype")
public class JPbSocketNettyClient extends BaseClient {
    private static Logger logger = LoggerFactory.getLogger(JPbSocketNettyClient.class);
    private String ip;
    private int port;
    private ChannelHandlerContext ctx;
    //@Resource
    private BlockingQueue<Msg> blockingQueue = new LinkedBlockingQueue<Msg>();
    private Bootstrap bootstrap = new Bootstrap(); // (1)
    private JPbSocketNettyClientHandler jPbSocketNettyClientHandler = new JPbSocketNettyClientHandler();
    public JPbSocketNettyClient() {

    }

    @Override
    public void write(Msg msg) {
        logger.info("add:" + msg);
        blockingQueue.add(msg);
    }

    @Override
    public void init(String ip, int port) {
        //保留这二个参数，重连会用上
        this.ip = ip;
        this.port = port;
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {

            bootstrap.group(workerGroup); // (2)
            bootstrap.channel(NioSocketChannel.class); // (3)
            bootstrap.option(ChannelOption.SO_KEEPALIVE, true); // (4)
            bootstrap.handler(new ChannelInitializer<SocketChannel>() {
                @Override
                public void initChannel(SocketChannel ch) throws Exception {
                    ch.pipeline().addLast(new LoggingHandler());
                    ch.pipeline().addLast(new JPbEncoder());
                    ch.pipeline().addLast(new JPbDecoder());
                    ch.pipeline().addLast("handler", jPbSocketNettyClientHandler);
                }
            });
            startWorking();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private void connect() {
        ChannelFuture channelFuture = bootstrap.connect(ip, port); // (5)
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        if (jPbSocketNettyClientHandler.isConnect()) {
            ctx = jPbSocketNettyClientHandler.getCtx();
        }
        try {
            channelFuture.sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    private void startWorking() {
        new Thread(new Runnable() {
            public void run() {
                while (true) {
                    if (!jPbSocketNettyClientHandler.isConnect()) {
                        try {
                            connect();
                            if (!jPbSocketNettyClientHandler.isConnect()) {
                                continue;
                            }
                        } catch (Exception e) {
                            e.printStackTrace();
                            logger.error("链接出错，请确认目标服务器已启动");
                            continue;
                        }
                    }
                    try {
                        if (!jPbSocketNettyClientHandler.isConnect()) {
                            continue;
                        }
                        Msg msg = blockingQueue.take();
                        logger.info("take:" + msg);
                        ctx.writeAndFlush(msg);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        }).start();
    }
}
